package app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.WatchVideoBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DWSInteractionVideo extends BaseApp {
    public static void main(String[] args) {
        new DWSInteractionVideo().start(10025,4,"dwsvideo", Constant.DWD_INTER_VIDEO);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<WatchVideoBean> reslutDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(s)) {
                            JSONObject jsonObject = JSON.parseObject(s);
                            collector.collect(jsonObject);
                        }
                    }
                }
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                }
                        )
        ).keyBy(jsonObject -> jsonObject.getString("sid")).map(
                new MapFunction<JSONObject, WatchVideoBean>() {
                    @Override
                    public WatchVideoBean map(JSONObject jsonObject) throws Exception {
                        Long play_sec = jsonObject.getLong("play_sec");
                        String video_id = jsonObject.getString("video_id");
                        return new WatchVideoBean("", "", "", video_id, "", 0L, 0L, play_sec);
                    }
                }
        ).windowAll(
                // TumblingProcessingTimeWindows.of(Time.seconds(5))
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).reduce(
                new ReduceFunction<WatchVideoBean>() {
                    @Override
                    public WatchVideoBean reduce(WatchVideoBean v1, WatchVideoBean v2) throws Exception {
                        v1.setPlay_time(v1.getPlay_time() > v2.getPlay_time() ? v1.getPlay_time() : v2.getPlay_time());
                        return v1;
                    }
                },
                new AllWindowFunction<WatchVideoBean, WatchVideoBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<WatchVideoBean> iterable, Collector<WatchVideoBean> collector) throws Exception {
                        WatchVideoBean bean = iterable.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(bean);
                    }
                }
        ).map(
                new MapFunction<WatchVideoBean, WatchVideoBean>() {
                    @Override
                    public WatchVideoBean map(WatchVideoBean watchVideoBean) throws Exception {
                        watchVideoBean.setPlay_count(1L);
                        watchVideoBean.setUser_count(1L);

                        return watchVideoBean;
                    }
                }
        ).keyBy(WatchVideoBean::getVideo_id).window(
                TumblingProcessingTimeWindows.of(Time.seconds(5))
                //TumblingEventTimeWindows.of(Time.seconds(5))
        ).reduce(
                new ReduceFunction<WatchVideoBean>() {
                    @Override
                    public WatchVideoBean reduce(WatchVideoBean v1, WatchVideoBean v2) throws Exception {
                        v1.setUser_count(v1.getUser_count() + v2.getUser_count());
                        v1.setPlay_count(v1.getPlay_count() + v2.getPlay_count());
                        v1.setPlay_time(v1.getPlay_time() + v2.getPlay_time());
                        return v1;
                    }
                }
        );
        AsyncDataStream.unorderedWait(
                reslutDS,
                new DimAsyncFunction<WatchVideoBean>(){
                    @Override
                    protected void addDim(WatchVideoBean obj, JSONObject dimJsonObj) {
                        obj.setChapter_name(dimJsonObj.getString("chapter_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_chapter_info";
                    }

                    @Override
                    public String getRowKey(WatchVideoBean obj) {
                        String video_id = obj.getVideo_id();
                        long l = Long.parseLong(video_id) + 20895;
                        return l+"";
                    }
                },60, TimeUnit.SECONDS
        ).map(new BeanToJsonStrMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_interaction_video_watch_window"));
    }
}
